package edu.hhu.wifispark.task.sparktask;

import java.io.Serializable;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import edu.hhu.wifispark.entry.Customer;
import edu.hhu.wifispark.entry.Record;
import edu.hhu.wifispark.util.PropUtil;
import edu.hhu.wifispark.util.SparkProUtil;
import scala.Tuple2;

public class CustomerTask extends Thread implements Serializable {
	private static final long serialVersionUID = 1L;
	public List<Record> recordInfo;
	public List<Record> recordPeri;
	String taskName = null;
	long stime, etime;
	public CustomerTask(){
		this.taskName = "customer";
	}
	public CustomerTask(long time){
		this.taskName = "customer" + time;
	}
	public CustomerTask(long starttime, long endtime){
		this.taskName = "customer" + starttime;
		this.stime = starttime;
		this.etime = endtime;
	}
	
	
	@SuppressWarnings("resource")
	public void run(){
		SparkConf sparkConf = new SparkConf().setAppName(taskName);
		sparkConf.setMaster(SparkProUtil.MASTER);
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<Record> infoRDD = jsc.parallelize(recordInfo);
		JavaRDD<Record> periRDD = jsc.parallelize(recordPeri);
		
		JavaRDD<Record> recordRDD = infoRDD.union(periRDD);
		JavaPairRDD<String, Record> recordMap = recordRDD.mapToPair(new PairFunction<Record, String, Record>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Tuple2<String, Record> call(Record t) throws Exception {
				return new Tuple2<String, Record>(t.getStr("mac"), t);
			}
		});
		JavaPairRDD<String, Record> recordPairRDD = recordMap.reduceByKey(new Function2<Record, Record, Record>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Record call(Record v1, Record v2) throws Exception {
				if("000".equals(v1.getStr("peri"))){
					v1.peri = v2.getStr("peri");
					return v1;
				}else{
					v2.peri= v1.getStr("peri");
					return v2;
				}
			}
		});
		
		JavaRDD<Customer> customerRDD = recordPairRDD.map(new Function<Tuple2<String, Record>, Customer>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Customer call(Tuple2<String, Record> record) throws Exception {
				Customer customer = new Customer();
				Record v1 = record._2();
				if(v1.peri == null){
					return null;
				}
				int freq = v1.getLong("counts").intValue();
				int avetime = v1.getBigDecimal("sums").intValue()/freq;
				String mac = v1.getStr("mac");
				String[] times = v1.peri.split(",");
				long period = -1;
				if(times.length == 2){
					period = Long.valueOf(times[0]) - Long.valueOf(times[1]);
				}
				customer.set("mac", mac);
				customer.set("atime", stime);
				customer.set("period", period);
				customer.set("freq", freq);
				customer.set("avetime", avetime);
				if(avetime < PropUtil.OUTOUT_LINE){
					customer.set("inin", 0);
					customer.set("outout", 1);
				}
				else if(avetime < PropUtil.ININ_LINE){
					customer.set("inin", 0);
					customer.set("outout", 0);
				}
				else{
					customer.set("inin", 1);
					customer.set("outout", 0);
				}
				return customer;
			}
		});
		List<Customer> customers = customerRDD.collect();
		jsc.stop();
		for(Customer customer : customers){
			if(customer != null)
				customer.save();
		}
	}
	
	public void setRecords(List<Record> info,List<Record> peri){
		this.recordInfo = info;
		this.recordPeri = peri;
	}
}
